package business.server;

import business.message.NetworkMessageHandler;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import demo.core.RaftNode;
import demo.core.RaftState;
import demo.message.RaftMessage;

import java.io.*;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * Raft键值存储HTTP服务器
 *
 * <p>将Raft分布式一致性算法实现的键值存储通过HTTP API对外提供服务。
 * 这是一个生产就绪的示例，展示了如何将Raft算法集成到实际业务系统中。</p>
 *
 * <p><b>支持的API接口：</b></p>
 * <ul>
 *   <li><b>PUT /kv/{key}</b>：设置键值对（写操作，需要通过Leader）</li>
 *   <li><b>GET /kv/{key}</b>：获取键的值（读操作，可以从任意节点读取）</li>
 *   <li><b>DELETE /kv/{key}</b>：删除键值对（写操作，需要通过Leader）</li>
 *   <li><b>GET /status</b>：获取节点状态信息</li>
 *   <li><b>GET /health</b>：健康检查接口</li>
 *   <li><b>GET /metrics</b>：监控指标接口</li>
 * </ul>
 *
 * <p><b>高可用特性：</b></p>
 * <ul>
 *   <li>自动Leader重定向：写请求会自动重定向到当前Leader</li>
 *   <li>只读优化：读请求可以从任意节点服务，提高性能</li>
 *   <li>健康检查：提供节点健康状态监控</li>
 *   <li>优雅关闭：支持优雅关闭和资源清理</li>
 * </ul>
 *
 * @author Raft Implementation Team
 * @version 1.0
 * @since 1.0
 */
public class RaftKVServer {

    private final RaftNode raftNode;
    private final KeyValueStateMachine stateMachine;
    private final HttpServer httpServer;
    private final int port;
    private final ExecutorService executor;
    private final Map<String, String> peerAddresses; // nodeId -> address映射

    // 共享的消息处理器实例

    /**
     * 构造Raft KV服务器
     *
     * @param nodeId        节点ID
     * @param port          HTTP服务端口
     * @param peers         其他节点ID列表
     * @param peerAddresses 节点ID到地址的映射
     * @throws IOException 如果创建HTTP服务器失败
     */
    public RaftKVServer(String nodeId, int port, List<String> peers, Map<String, String> peerAddresses) throws IOException {
        this.port = port;
        this.peerAddresses = new HashMap<>(peerAddresses);
        this.executor = Executors.newFixedThreadPool(10);

        // 创建状态机
        this.stateMachine = new KeyValueStateMachine();

        // 创建Raft节点，使用网络消息处理器
        NetworkMessageHandler messageHandler = new NetworkMessageHandler(nodeId, this.peerAddresses);
        this.raftNode = new RaftNode(nodeId, peers, messageHandler);

        // 创建HTTP服务器
        this.httpServer = HttpServer.create(new InetSocketAddress(port), 0);
        setupHttpHandlers();
        this.httpServer.setExecutor(executor);

        System.out.println("Raft KV服务器 " + nodeId + " 在端口 " + port + " 启动");
    }

    /**
     * 设置HTTP处理器
     */
    private void setupHttpHandlers() {
        // 键值操作API
        httpServer.createContext("/kv/", new KVHandler());

        // 状态查询API
        httpServer.createContext("/status", new StatusHandler());

        // 健康检查API
        httpServer.createContext("/health", new HealthHandler());

        // 监控指标API
        httpServer.createContext("/metrics", new MetricsHandler());

        // 集群信息API
        httpServer.createContext("/cluster", new ClusterHandler());

        // Raft消息处理API
        httpServer.createContext("/raft/vote", new VoteRequestHandler());
        httpServer.createContext("/raft/append", new AppendEntriesHandler());
    }

    /**
     * Raft投票请求处理器
     */
    private class VoteRequestHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            if (!"POST".equals(exchange.getRequestMethod())) {
                sendResponse(exchange, 405, "Method not allowed");
                return;
            }

            try {
                // 读取请求体
                ObjectInputStream in = new ObjectInputStream(exchange.getRequestBody());
                RaftMessage.VoteRequest request = (RaftMessage.VoteRequest) in.readObject();

                // 处理请求
                RaftMessage.VoteResponse response = raftNode.handleVoteRequest(request);

                // 发送响应
                exchange.getResponseHeaders().set("Content-Type", "application/octet-stream");
                exchange.sendResponseHeaders(200, 0);
                ObjectOutputStream out = new ObjectOutputStream(exchange.getResponseBody());
                out.writeObject(response);
                out.flush();
            } catch (Exception e) {
                System.err.println("处理投票请求失败: " + e.getMessage());
                // 创建一个错误响应，包含必要的信息
                try {
                    ObjectInputStream in = new ObjectInputStream(exchange.getRequestBody());
                    RaftMessage.VoteRequest request = (RaftMessage.VoteRequest) in.readObject();
                    RaftMessage.VoteResponse errorResponse = new RaftMessage.VoteResponse(
                        request.getTerm(), raftNode.getNodeId(), request.getCandidateId(), false);
                    
                    exchange.getResponseHeaders().set("Content-Type", "application/octet-stream");
                    exchange.sendResponseHeaders(200, 0);
                    ObjectOutputStream out = new ObjectOutputStream(exchange.getResponseBody());
                    out.writeObject(errorResponse);
                    out.flush();
                } catch (Exception ex) {
                    sendResponse(exchange, 500, "Internal server error");
                }
            } finally {
                exchange.close();
            }
        }
    }

    /**
     * Raft附加日志请求处理器
     */
    private class AppendEntriesHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            if (!"POST".equals(exchange.getRequestMethod())) {
                sendResponse(exchange, 405, "Method not allowed");
                return;
            }

            try {
                // 读取请求体
                ObjectInputStream in = new ObjectInputStream(exchange.getRequestBody());
                RaftMessage.AppendEntriesRequest request = (RaftMessage.AppendEntriesRequest) in.readObject();

                // 处理请求
                RaftMessage.AppendEntriesResponse response = raftNode.handleAppendEntries(request);

                // 发送响应
                exchange.getResponseHeaders().set("Content-Type", "application/octet-stream");
                exchange.sendResponseHeaders(200, 0);
                ObjectOutputStream out = new ObjectOutputStream(exchange.getResponseBody());
                out.writeObject(response);
                out.flush();
            } catch (Exception e) {
                System.err.println("处理附加日志请求失败: " + e.getMessage());
                sendResponse(exchange, 500, "Internal server error");
            } finally {
                exchange.close();
            }
        }
    }

    /**
     * 获取节点ID
     */
    public String getNodeId() {
        return raftNode.getNodeId();
    }

    /**
     * 获取节点地址
     */
    public String getAddress() {
        return peerAddresses.get(getNodeId());
    }

    /**
     * 启动服务器
     */
    public void start() {
        httpServer.start();
        System.out.println("HTTP服务器在端口 " + port + " 启动成功");
    }

    /**
     * 停止服务器
     */
    public void stop() {
        System.out.println("正在关闭Raft KV服务器...");

        // 停止HTTP服务器
        httpServer.stop(5);

        // 关闭线程池
        executor.shutdown();
        try {
            if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        // 关闭Raft节点
        raftNode.gracefulShutdown();

        System.out.println("Raft KV服务器已关闭");
    }

    /**
     * 键值操作处理器
     */
    private class KVHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            String method = exchange.getRequestMethod();
            String path = exchange.getRequestURI().getPath();

            try {
                // 解析键名
                String key = extractKeyFromPath(path);
                if (key == null) {
                    sendResponse(exchange, 400, "Invalid key format");
                    return;
                }

                switch (method) {
                    case "GET":
                        handleGet(exchange, key);
                        break;
                    case "PUT":
                        handlePut(exchange, key);
                        break;
                    case "DELETE":
                        handleDelete(exchange, key);
                        break;
                    default:
                        sendResponse(exchange, 405, "Method not allowed");
                }
            } catch (Exception e) {
                sendResponse(exchange, 500, "Internal server error: " + e.getMessage());
            }
        }

        private String extractKeyFromPath(String path) {
            // /kv/key -> key
            if (path.startsWith("/kv/") && path.length() > 4) {
                return path.substring(4);
            }
            return null;
        }

        private void handleGet(HttpExchange exchange, String key) throws IOException {
            // 读操作可以从本地状态机直接读取
            String value = stateMachine.getValue(key);
            if (value != null) {
                sendResponse(exchange, 200, value);
            } else {
                sendResponse(exchange, 404, "Key not found");
            }
        }

        private void handlePut(HttpExchange exchange, String key) throws IOException {
            // 读取请求体作为值
            String value = readRequestBody(exchange);
            if (value == null || value.trim().isEmpty()) {
                sendResponse(exchange, 400, "Value is required");
                return;
            }

            // 写操作需要通过Raft协议
            if (raftNode.getState() != RaftState.LEADER) {
                // 如果不是Leader，返回重定向信息
                String leaderAddress = findLeaderAddress();
                if (leaderAddress != null) {
                    exchange.getResponseHeaders().set("Location", "http://" + leaderAddress + exchange.getRequestURI().toString());
                    sendResponse(exchange, 307, "Redirect to leader: " + leaderAddress);
                } else {
                    sendResponse(exchange, 503, "No leader available");
                }
                return;
            }

            // 通过Raft提交命令
            String command = "SET " + key + " " + value;
            boolean success = raftNode.appendEntry(command);

            if (success) {
                // 等待命令被应用（简化实现，实际应该有超时机制）
                waitForCommandApplication(command);
                sendResponse(exchange, 200, "OK");
            } else {
                sendResponse(exchange, 503, "Failed to commit command");
            }
        }

        private void handleDelete(HttpExchange exchange, String key) throws IOException {
            // 写操作需要通过Raft协议
            if (raftNode.getState() != RaftState.LEADER) {
                String leaderAddress = findLeaderAddress();
                if (leaderAddress != null) {
                    exchange.getResponseHeaders().set("Location", "http://" + leaderAddress + exchange.getRequestURI().toString());
                    sendResponse(exchange, 307, "Redirect to leader: " + leaderAddress);
                } else {
                    sendResponse(exchange, 503, "No leader available");
                }
                return;
            }

            String command = "DELETE " + key;
            boolean success = raftNode.appendEntry(command);

            if (success) {
                waitForCommandApplication(command);
                sendResponse(exchange, 200, "OK");
            } else {
                sendResponse(exchange, 503, "Failed to commit command");
            }
        }

        private String readRequestBody(HttpExchange exchange) throws IOException {
            try (BufferedReader reader = new BufferedReader(
                    new InputStreamReader(exchange.getRequestBody(), StandardCharsets.UTF_8))) {
                StringBuilder sb = new StringBuilder();
                String line;
                while ((line = reader.readLine()) != null) {
                    sb.append(line);
                }
                return sb.toString();
            }
        }

        private void waitForCommandApplication(String command) {
            // 简化的等待逻辑，实际应该更精确
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    /**
     * 状态查询处理器
     */
    private class StatusHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            if (!"GET".equals(exchange.getRequestMethod())) {
                sendResponse(exchange, 405, "Method not allowed");
                return;
            }

            Map<String, Object> status = new HashMap<>();
            status.put("nodeId", raftNode.getNodeId());
            status.put("state", raftNode.getState().toString());
            status.put("term", raftNode.getCurrentTerm());
            status.put("commitIndex", raftNode.getCommitIndex());
            status.put("logSize", raftNode.getLogSize());
            status.put("stateMachine", stateMachine.getState());

            String json = mapToJson(status);
            sendJsonResponse(exchange, 200, json);
        }
    }

    /**
     * 健康检查处理器
     */
    private class HealthHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            if (!"GET".equals(exchange.getRequestMethod())) {
                sendResponse(exchange, 405, "Method not allowed");
                return;
            }

            // 简单的健康检查：检查Raft节点是否正常
            boolean healthy = raftNode.getState() != null;

            Map<String, Object> health = new HashMap<>();
            health.put("status", healthy ? "UP" : "DOWN");
            health.put("timestamp", System.currentTimeMillis());

            String json = mapToJson(health);
            int statusCode = healthy ? 200 : 503;
            sendJsonResponse(exchange, statusCode, json);
        }
    }

    /**
     * 监控指标处理器
     */
    private class MetricsHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            if (!"GET".equals(exchange.getRequestMethod())) {
                sendResponse(exchange, 405, "Method not allowed");
                return;
            }

            Map<String, Object> metrics = new HashMap<>();
            metrics.put("raft_term", raftNode.getCurrentTerm());
            metrics.put("raft_log_size", raftNode.getLogSize());
            metrics.put("raft_commit_index", raftNode.getCommitIndex());
            metrics.put("kv_store_size", stateMachine.size());
            metrics.put("memory_usage", Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory());
            metrics.put("uptime", System.currentTimeMillis());

            String json = mapToJson(metrics);
            sendJsonResponse(exchange, 200, json);
        }
    }

    /**
     * 集群信息处理器
     */
    private class ClusterHandler implements HttpHandler {
        @Override
        public void handle(HttpExchange exchange) throws IOException {
            if (!"GET".equals(exchange.getRequestMethod())) {
                sendResponse(exchange, 405, "Method not allowed");
                return;
            }

            Map<String, Object> cluster = new HashMap<>();
            cluster.put("leader", findLeaderAddress());
            cluster.put("nodes", peerAddresses);
            cluster.put("clusterSize", peerAddresses.size());

            String json = mapToJson(cluster);
            sendJsonResponse(exchange, 200, json);
        }
    }

    /**
     * 查找Leader地址
     */
    private String findLeaderAddress() {
        // 在实际实现中，这里需要查询其他节点的状态
        // 当前简化实现，只返回本节点地址（如果是Leader）
        if (raftNode.getState() == RaftState.LEADER) {
            return "localhost:" + port;
        }
        return null; // 在实际实现中应该查询其他节点
    }

    /**
     * 发送普通响应
     */
    private void sendResponse(HttpExchange exchange, int statusCode, String response) throws IOException {
        byte[] bytes = response.getBytes(StandardCharsets.UTF_8);
        exchange.sendResponseHeaders(statusCode, bytes.length);
        try (OutputStream os = exchange.getResponseBody()) {
            os.write(bytes);
        }
    }

    /**
     * 发送JSON响应
     */
    private void sendJsonResponse(HttpExchange exchange, int statusCode, String json) throws IOException {
        exchange.getResponseHeaders().set("Content-Type", "application/json");
        sendResponse(exchange, statusCode, json);
    }

    /**
     * 简单的Map到JSON转换（生产环境建议使用专业的JSON库）
     */
    private String mapToJson(Map<String, Object> map) {
        StringBuilder sb = new StringBuilder();
        sb.append("{");
        boolean first = true;
        for (Map.Entry<String, Object> entry : map.entrySet()) {
            if (!first) sb.append(",");
            sb.append("\"").append(entry.getKey()).append("\":");
            Object value = entry.getValue();
            if (value instanceof String) {
                sb.append("\"").append(value).append("\"");
            } else if (value instanceof Map) {
                sb.append(mapToJson((Map<String, Object>) value));
            } else if (value instanceof List) {
                sb.append(listToJson((List<?>) value));
            } else {
                sb.append(value);
            }
            first = false;
        }
        sb.append("}");
        return sb.toString();
    }

    /**
     * 简单的List到JSON转换
     */
    private String listToJson(List<?> list) {
        StringBuilder sb = new StringBuilder();
        sb.append("[");
        boolean first = true;
        for (Object item : list) {
            if (!first) sb.append(",");
            if (item instanceof String) {
                sb.append("\"").append(item).append("\"");
            } else {
                sb.append(item);
            }
            first = false;
        }
        sb.append("]");
        return sb.toString();
    }

    /**
     * 主方法 - 演示如何启动一个Raft KV集群
     */
        public static void main(String[] args) throws Exception {
        if (args.length < 3) {
            System.out.println("用法: java RaftKVServer <nodeId> <port> <clusterConfig>");
            System.out.println("示例: java RaftKVServer node1 8081 node1:8081,node2:8082,node3:8083");
            System.exit(1);
        }
        
        String nodeId = args[0];
        int port = Integer.parseInt(args[1]);
        String clusterConfig = args[2];
        
        // 解析集群配置
        List<String> peers = new ArrayList<>();
        Map<String, String> peerAddresses = new HashMap<>();
        
        String[] nodes = clusterConfig.split(",");
        for (String node : nodes) {
            String[] parts = node.split(":");
            if (parts.length == 2) {
                String peerId = parts[0];
                if (!peerId.equals(nodeId)) {  // 不把自己加入到peers列表中
                    peers.add(peerId);
                }
                peerAddresses.put(peerId, "localhost:" + parts[1]);
            }
        }
        
        System.out.println("节点 " + nodeId + " 的配置:");
        System.out.println("- 监听端口: " + port);
        System.out.println("- 其他节点: " + peers);
        System.out.println("- 地址映射: " + peerAddresses);

        // 创建并启动服务器
        RaftKVServer server = new RaftKVServer(nodeId, port, peers, peerAddresses);
        server.start();

        // 添加关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(server::stop));

        System.out.println("服务器启动完成，按 Ctrl+C 停止");

        // 保持运行
        Thread.currentThread().join();
    }
} 